package org.fastsyncer.framework.manager;

import java.io.File;
import java.util.ArrayList;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedQueue;

import org.apache.commons.lang.StringUtils;
import org.fastsyncer.common.constant.CommonConstant;
import org.fastsyncer.common.constant.ConnectorConstant;
import org.fastsyncer.common.constant.MappingConstant;
import org.fastsyncer.common.constant.RestConstant;
import org.fastsyncer.common.entity.DatabaseConfig;
import org.fastsyncer.common.entity.Mapping;
import org.fastsyncer.common.entity.MappingTask;
import org.fastsyncer.common.util.ApplicationUtil;
import org.fastsyncer.common.util.DateUtil;
import org.fastsyncer.manager.framework.ManagerFactory;
import org.fastsyncer.manager.health.HeartBeat;
import org.fastsyncer.manager.store.DataCenter;
import org.fastsyncer.manager.util.ManagerUtil;
import org.json.JSONArray;
import org.json.JSONObject;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * 驱动业务操作API,包含驱动新增、修改、删除、查询、启动、停止
 * @author YangPei
 *
 */
public class ManagerService {

	private static Logger logger = LoggerFactory.getLogger(ManagerService.class);

	private static final ManagerService instance = new ManagerService();

	private ManagerService() {
	}

	public static ManagerService getInstance() {
		return instance;
	}

	private DataCenter data = DataCenter.getInstance();

	private ManagerFactory manager = ManagerFactory.getInstance();

	/**
	 * 新增驱动
	 * @Title: add 
	 * @Description: 新增驱动
	 * @param mapping
	 * @throws Exception
	 * @return: String 驱动ID
	 */
	public String add(String mapping) throws Exception {
		MappingTask mappingTask = ManagerUtil.parseMapping(mapping);
		boolean store = data.storeMapping(mappingTask.getId(), mappingTask);
		if (!store) {
			logger.error("Add driver failed!");
			throw new Exception("新增驱动失败!");
		}
		return mappingTask.getId();
	}

	/**
	 * 修改驱动
	 * 
	 * @param mapping
	 * @throws Exception
	 */
	public String update(String mapping) throws Exception {
		MappingTask mappingTask = ManagerUtil.parseMapping(mapping);
		// 当前为启用
		if (!mappingTask.isEnable()) {
			throw new Exception("驱动已被禁用.");
		}
		
		// 检查驱动是否在运行,如果运行给出提示
		String model = mappingTask.getModel();
		if(MappingConstant.SYNC_ALL.equals(model)){
		    //检查驱动是否处于运行状态
		    if(this.isRun(mappingTask.getId())){
		        throw new Exception("驱动正在运行,请先停止.");
		    }
		    
		    // 获取策略
            HashMap<String, HashMap<String, String>> policy = mappingTask.getPolicy();
            HashMap<String, String> po = policy.get(MappingConstant.POLICY_INCREMENT);
            // 关闭同步模式下驱动功能
            if (po != null) {
                po.put("enable", "false");
            }
		}else{
		    // 获取策略
		    HashMap<String, HashMap<String, String>> policy = mappingTask.getPolicy();
		    HashMap<String, String> po = policy.get(MappingConstant.POLICY_INCREMENT);
		    // 是否开启功能
		    if (po != null) {
		        if(Boolean.valueOf(po.get("enable"))){
		            throw new Exception("驱动正在运行,请先停止.");
		        }
		    }
		}
		
		// 保存驱动配置
		boolean store = data.storeMapping(mappingTask.getId(), mappingTask);
		if (!store) {
			logger.error("Update driver failed!");
			throw new Exception("修改驱动失败!");
		}
		
		// 标记当前驱动已发生变更
		ApplicationUtil.setKey(CommonConstant.DRIVER_CONFIG_UPDATE + mappingTask.getId(), "true");
		return "修改驱动成功!";
	}

	/**
	 * 恢复驱动
	 * 
	 * @param id
	 * @throws Exception
	 */
	public String enable(String id) throws Exception {
		MappingTask mappingTask = (MappingTask) data.getMapping(id);
		if(mappingTask==null){
			throw new Exception("驱动不存在.");
		}
		// 修改状态为false,去掉启用时间
		mappingTask.setEnable(true);
		mappingTask.setStartDate(new Date());

		boolean store = data.storeMapping(id, mappingTask);
		if (!store) {
			logger.error("Update driver failed!");
			throw new Exception("修改驱动状态失败!");
		}
		return "恢复驱动成功!";
	}
	
	/**
	 * 禁用驱动
	 * 
	 * @param name
	 * @throws Exception
	 */
	public String disable(String id) throws Exception {
		MappingTask mappingTask = (MappingTask) data.getMapping(id);
		if(mappingTask==null){
			throw new Exception("驱动不存在.");
		}
		//检查驱动是否处于运行状态
		if(this.isRun(id)){
			throw new Exception("请先停止驱动或等待任务结束.");
		}
		
		// 获取策略
        HashMap<String, HashMap<String, String>> policy = mappingTask.getPolicy();
        HashMap<String, String> po = policy.get(MappingConstant.POLICY_INCREMENT);
        // 是否开启功能
        if (po != null) {
            if(Boolean.valueOf(po.get("enable"))){
                throw new Exception("驱动正在运行,请先停止.");
            }
        }
		
		// 修改状态为true,设置最新启用时间
		mappingTask.setEnable(false);
		mappingTask.setStartDate(null);

		boolean store = data.storeMapping(id, mappingTask);
		if (!store) {
			logger.error("Update driver failed!");
			throw new Exception("修改驱动状态失败!");
		}
		return "禁用驱动成功!";
	}

	/**
	 * 启动驱动
	 * 
	 * @param driverId
	 * @throws Exception
	 */
	public String start(String driverId) throws Exception {
		MappingTask mappingTask = (MappingTask) data.getMapping(driverId);
		
		// 1、检查驱动
		if(mappingTask==null){
			throw new IllegalArgumentException("驱动不存在！");
		}

		// 2、检查是否启用
		if (!mappingTask.isEnable()) {
			throw new IllegalArgumentException("驱动状态必须为启用！");
		}
		
		// 3、如果上一次的任务已经执行完成，则还原同步默认值
		if (mappingTask.getProcess() == mappingTask.getProcessCompleted()) {
			mappingTask.setSuccess(0);
			mappingTask.setFail(0);
			mappingTask.setError(0);
			mappingTask.setProcessCompleted(0);
			mappingTask.setRunTime(0L);
		}
		
		// 分别处理数据迁移和数据同步业务
		String model = mappingTask.getModel();
		switch (model) {
		case MappingConstant.SYNC_ALL:
			// 4、检查同步属性、主键
			this.validateSyncFields(mappingTask);

			// 5、校验同步任务是否正在处理中
			manager.validateTask(driverId);
			// 6、处理任务
			manager.handle(mappingTask);
			break;
		case MappingConstant.SYNC_INCREMENT:
            // 检查同步属性
		    this.validateSyncFields(mappingTask);
		    
			HashMap<String, HashMap<String, String>> policy = mappingTask.getPolicy();
			HashMap<String, String> inc = policy.get(MappingConstant.POLICY_INCREMENT);
			if(inc == null){
			    policy.put(MappingConstant.POLICY_INCREMENT, new HashMap<String, String>());
			    inc = policy.get(MappingConstant.POLICY_INCREMENT);
			}
			inc.put("enable", "true");
			boolean store = data.storeMapping(driverId, mappingTask);
			if (!store) {
				logger.error("Start driver failed!");
				throw new Exception("启动驱动失败!");
			}
			break;
		default:
			break;
		}
		return "驱动启动成功！";
	}
	
	/**
	 * 停止驱动
	 * 
	 * @param mapping
	 * @throws Exception
	 */
	public String stop(String driverId) throws Exception {
		logger.info("Stopping the task.");
		MappingTask mappingTask = (MappingTask) data.getMapping(driverId);
		
		// 1、检查驱动
		if(mappingTask==null){
			throw new IllegalArgumentException("驱动不存在！");
		}
		
		// 2、检查是否启用
		if (!mappingTask.isEnable()) {
			throw new IllegalArgumentException("驱动状态必须为启用！");
		}
		
		String res = "停止驱动成功！";
		// 分别处理数据迁移和数据同步业务
		String model = mappingTask.getModel();
		switch (model) {
		case MappingConstant.SYNC_ALL:
			// 3、移除正在处理的同步任务
			boolean removeRunning = manager.removeRunning(driverId);
			boolean removeQueue = manager.removeQueue(driverId);
			res = removeRunning || removeQueue ? res : "没有驱动可以停止！";
			break;
		case MappingConstant.SYNC_INCREMENT:
			HashMap<String, HashMap<String, String>> policy = mappingTask.getPolicy();
			HashMap<String, String> inc = policy.get(MappingConstant.POLICY_INCREMENT);
			if(inc == null){
                policy.put(MappingConstant.POLICY_INCREMENT, new HashMap<String, String>());
                inc = policy.get(MappingConstant.POLICY_INCREMENT);
            }
			inc.put("enable", "false");
			boolean store = data.storeMapping(driverId, mappingTask);
			if (!store) {
				logger.error("Stop driver failed!");
				throw new Exception("停止驱动失败!");
			}
			break;
		default:
			break;
		}
		return res;
	}

	/**
	 * 删除驱动
	 * 
	 * @param id
	 * @throws Exception
	 */
	public String delete(String id) throws Exception {
		MappingTask mappingTask = (MappingTask) data.getMapping(id);
		if (mappingTask.isEnable()) {
			throw new Exception("请先禁用驱动!");
		}
		// 移除正在运行的任务
		manager.removeRunning(id);
		manager.removeQueue(id);
		boolean remove = data.remove(mappingTask.getId());
		if (!remove) {
			logger.error("Delete driver failed!");
			throw new Exception("删除驱动失败!");
		}
		return "删除驱动成功!";
	}
	
	/**
	 * 获取驱动
	 * @param dirverId
	 * @return
	 * @throws Exception
	 */
	public Object getDriver(String dirverId) throws Exception {
		return data.getMapping(dirverId);
	}

	/**
	 * 获取所有驱动
	 * @return
	 * @throws Exception
	 */
	public List<Object> getDriverAll() throws Exception {
		List<Object> list = data.getMappingList();
		for (Object object : list) {
			MappingTask task = (MappingTask) object;

			Date startDate = task.getStartDate();
			task.setStartDateStr("0天0小时0分0秒");
			if (startDate != null) {
				// 获取时间差
				task.setStartDateStr(DateUtil.getDiff(startDate, new Date()));
			}

			// 检测驱动可用性
			this.isAlive(task);

			// 获取驱动运行状态
			this.getDriverRunState(task);
		}
		return list;
	}
	
	/**
	 * 获取驱动运行状态
	 * @Title: getDriverRunState 
	 * @Description: 获取驱动运行状态
	 * @param task
	 * @return: void
	 */
	private void getDriverRunState(MappingTask task){
		// 分别处理数据迁移和数据同步业务
		boolean run = false;
		switch (task.getModel()) {
		case MappingConstant.SYNC_ALL:
			// 获取运行状态
			run = this.isRun(task.getId());
			break;
		case MappingConstant.SYNC_INCREMENT:
			// 获取策略
			HashMap<String, HashMap<String, String>> policy = task.getPolicy();
			HashMap<String, String> po = policy.get(MappingConstant.POLICY_INCREMENT);
			// 是否开启功能
			if (po != null) {
				String enable = po.get("enable");
				run = enable!=null && Boolean.valueOf(enable);
			}
			break;
		default:
			break;
		}
		task.setRun(run);
	}
	
	/**
	 * 获取驱动同步日志
	 * 
	 * @param id
	 * @throws Exception
	 */
	public File getMappingLog(String id) {
		return data.getMappingLog(id);
	}

	/**
	 * 获取正在排队的同步任务
	 * 
	 * @return
	 */
	public List<String> getQueue() {
		List<String> list = new ArrayList<String>();
		ConcurrentLinkedQueue<String> queue = manager.getQueue();
		for (String t : queue) {
			list.add(t);
		}
		return list;
	}

	/**
	 * 获取正在运行的同步任务
	 * 
	 * @return
	 */
	public List<String> getRunning() {
		List<String> list = new ArrayList<String>();
		ConcurrentHashMap<String, String> running = manager.getRunning();
		for (Map.Entry<String, String> entry : running.entrySet()) {
			list.add(entry.getKey());
		}
		return list;
	}
	
	/**
	 * 
	 * 查询是否在运行
	 * @param id 驱动ID
	 * @return true:正在运行，否则false
	 */
	private boolean isRun(String id) {
		//判断是否在排队
		ConcurrentLinkedQueue<String> queue = manager.getQueue();
		if(queue.contains(id)){
			return true;
		}
		//判断是否在运行
		ConcurrentHashMap<String, String> running = manager.getRunning();
		return running.get(id)!=null;
	}
	
	/**
	 * 
	 * 查询驱动是否可用
	 * @param name 驱动名称
	 * @return true:正在运行，否则false
	 */
	private void isAlive(MappingTask task) {
		try {
			String id = task.getId();
			HashMap<String, Object> driver = HeartBeat.getInstance().getDriver(id);
			if(driver==null){
				return;
			}
			
			// 数据源是否可用
			boolean sourceAlive = (boolean) driver.get(CommonConstant.DRIVER_CONFIG_STATE_SOURCE);
			task.getSourceMapping().setAlive(sourceAlive);
			
			// 目标源是否可用
			@SuppressWarnings("unchecked")
			List<Boolean> targetAlives = (List<Boolean>) driver.get(CommonConstant.DRIVER_CONFIG_STATE_TARGET);
			if(targetAlives==null){
				return;
			}
			List<Mapping> mappings = task.getMappings();
			int mps = mappings.size();
			int size = targetAlives.size();
			if(mps!=size){
				return;
			}
			for (int i = 0; i < size; i++) {
				mappings.get(i).setAlive(targetAlives.get(i));
			}
		} catch (Exception e) {
			logger.error(e.getMessage());
		}
	}
	
	/**
	 * 检查同步属性、主键
	 * @param mappingTask
	 * @throws IllegalArgumentException
	 */
	private void validateSyncFields(MappingTask mappingTask) throws IllegalArgumentException {
		Mapping sourceMapping = mappingTask.getSourceMapping();
		List<HashMap<String, Object>> sourceFileds = sourceMapping.getFileds();
		if (sourceFileds == null || sourceFileds.size() < 1) {
			throw new IllegalArgumentException("缺少基本信息!");
		}
		
		// 数据源连接器类型
		String sourceConn = sourceMapping.getConnector();
		// 目标源驱动集合
		List<Mapping> targetMappings = mappingTask.getMappings();
		for (Mapping map : targetMappings) {
			List<HashMap<String, Object>> fileds = map.getFileds();
			if (fileds == null || fileds.size() < 1) {
				throw new IllegalArgumentException("缺少基本信息!");
			}

			// 如果都是数据库连接器类型
			if (ConnectorConstant.DB_CONNECTOR.equals(sourceConn) && sourceConn.equals(map.getConnector())) {
				DatabaseConfig sourceConfig = (DatabaseConfig) sourceMapping.getConfig();
				DatabaseConfig targetConfig = (DatabaseConfig) map.getConfig();
				String sKey = sourceConfig.getPrimaryKey();
				String tKey = targetConfig.getPrimaryKey();

				// 检查数据源主键是否存在
				boolean isExistSourcePK = false;
				for (HashMap<String, Object> source : sourceFileds) {
					String sName = (String) source.get("name");
					if (sKey.equals(sName)) {
						isExistSourcePK = true;
						break;
					}
				}
				if (!isExistSourcePK) {
					throw new IllegalArgumentException("基本信息数据源缺少主键!");
				}
				// 检查目标源主键是否存在
				boolean isExistTargetPK = false;
				for (HashMap<String, Object> source : fileds) {
					String tName = (String) source.get("name");
					if (tKey.equals(tName)) {
						isExistTargetPK = true;
						break;
					}
				}
				if (!isExistTargetPK) {
					throw new IllegalArgumentException("基本信息目标源缺少主键!");
				}
			}
		}
		// END
	}

	/**
	 * 同步数据
	 * @Title: sync 
	 * @Description: 同步数据
	 * @param json 同步的內容,包括对应的驱动名,增刪改操作事件.
	 * @throws Exception
	 * @return: void
	 */
	public void sync(String json) throws Exception{
		// 消息正确性校验,将同步消息转换为JSON格式Object
		JSONObject handle = new JSONObject(json);
		// 获取驱动ID
		String taskId = handle.getString(RestConstant.TASKID);
		
		// 检查驱动是否存在
		MappingTask driver = (MappingTask) getDriver(taskId);
		if(null==driver){
            throw new Exception("驱动:" + taskId + "不存在.");
        }
		// 检查驱动是否启动
		this.getDriverRunState(driver);
		if(!driver.isRun()){
		    throw new Exception("驱动:" + taskId + "未启动.");
		}
		
		// 获取同步数据集合
		JSONArray msgArr = handle.getJSONArray(RestConstant.MSG);
		int msgArrLen = msgArr.length();
		if(msgArrLen < 1){
			throw new Exception("同步的数据为空.");
		}
		
		// 重新组装同步的数据格式
		JSONArray msg = new JSONArray();
		for (int i = 0; i < msgArrLen; i++) {
			JSONObject d = msgArr.getJSONObject(i);
			
			// 事件类型
			String eventType = d.getString(RestConstant.EVENTTYPE);
			JSONArray data = d.getJSONArray(RestConstant.DATA);
			JSONObject rObj = new JSONObject();
			rObj.put("eventType", eventType);
			
			//如果是删除操作
			if(StringUtils.equals(ConnectorConstant.OPERTION_DELETE, eventType)){
				rObj.put("before", data);
				rObj.put("after", new JSONArray());
			}else{
				//如果是新增、修改操作
				rObj.put("before", new JSONArray());
				rObj.put("after", data);
			}
			msg.put(rObj);
		}
		
		// 数据格式转换
		JSONObject sync = new JSONObject();
		sync.put("taskId", taskId);
		sync.put("msg", msg);
		
		// 通知manager处理同步消息 
		manager.handleSync(sync);
	}
	
}
